package detector

import (
    "github.com/samuel/go-zookeeper/zk"
    "sync"
    "time"
    "../common"
    "fmt"
)

//.h /usr/local/include
//.so /usr/local/include
//import "C"

type ZkDetector struct {
    activeMetaNode       common.Node
    myNode               common.Node
    serverMode           bool
    mu                   sync.RWMutex
    stopping             chan interface{}
    stopped              chan interface{}
    onSetActiveCallback  func(bool)
    onMetaLossCallback   func()
    onMetaChangeCallback func(common.Node) //rep should know who is meta
    zkConn               *zk.Conn
    opt                  ZkOption
    zkEvent              <-chan zk.Event
}

func (md *ZkDetector) tryToBeActive() bool {
    my := string(md.myNode)
    _, err := md.zkConn.Create(md.opt.ZkMetaPath, []byte(my), zk.FlagEphemeral, zk.WorldACL(zk.PermRead))
    if err == nil {
        md.mu.Lock()
        defer md.mu.Unlock()
        md.activeMetaNode = md.myNode
        return true
    }
    return false
}

func (md *ZkDetector) resetActiveMeta() {
    md.mu.Lock()
    defer md.mu.Unlock()
    md.activeMetaNode = ""
    md.onMetaLossCallback()
}

func (md *ZkDetector) loadNewActiveMeta() {
    fmt.Println("loadNewActiveMeta")
    data, _, err := md.zkConn.Get(md.opt.ZkMetaPath)
    md.mu.Lock()
    defer md.mu.Unlock()
    if err != nil {
        fmt.Println("found no meta registed")
        md.activeMetaNode = ""
        md.onMetaLossCallback()
    } else if metaNode := string(data); string(md.activeMetaNode) != metaNode {
        md.activeMetaNode = common.Node(metaNode)
        md.onMetaChangeCallback(common.Node(metaNode))
    } else {
        fmt.Println("do nothing")
    }
}

func (md *ZkDetector) IsActive() bool {
    return md.GetActiveMeta(false) == md.myNode
}

func (md *ZkDetector) GetActiveMeta(refresh bool) common.Node {
    if refresh {
        md.loadNewActiveMeta()
    }
    md.mu.RLock()
    defer md.mu.RUnlock()
    return md.activeMetaNode
}

func (md *ZkDetector) Stop() {
    close(md.stopping)
    <-md.stopped
}

func (md *ZkDetector) events() <-chan zk.Event {
    md.mu.RLock()
    defer md.mu.RUnlock()
    return md.zkEvent
}

func (md *ZkDetector) resetWatch() error {
    _, _, eventCh, err := md.zkConn.GetW(md.opt.ZkMetaPath)
    md.mu.Lock()
    defer md.mu.Unlock()
    md.zkEvent = nil
    if err != nil {
        return err
    }
    md.zkEvent = eventCh
    return nil
}

func NewZkDetectorOnRep(myNode common.Node, opt ZkOption,
    onMetaLossCallback func(), onMetaChangeCallback func(common.Node)) *ZkDetector {
    md := &ZkDetector{
        myNode:               myNode,
        serverMode:           false,
        stopping:             make(chan interface{}),
        stopped:              make(chan interface{}),
        onMetaLossCallback:   onMetaLossCallback,
        onMetaChangeCallback: onMetaChangeCallback,
        opt:                  opt,
    }
    return md
}

func NewZkDetectorOnMeta(myNode common.Node, opt ZkOption,
    onSetActiveCallback func(bool)) *ZkDetector {
    md := &ZkDetector{
        myNode:              myNode,
        serverMode:          true,
        stopping:            make(chan interface{}),
        stopped:             make(chan interface{}),
        onSetActiveCallback: onSetActiveCallback,
        opt:                 opt,
    }
    return md
}

func (md *ZkDetector) Start() error {
    zkConn, _, err := zk.Connect(md.opt.ZkAddresses, time.Duration(md.opt.ZkTimeoutMilliseconds)*time.Millisecond)
    if err != nil {
        return err
    }
    md.zkConn = zkConn
    md.resetWatch()
    tk := time.Tick(time.Duration(md.opt.ZkTimeoutMilliseconds) * time.Millisecond)
LOOP:
    for {
        select {
        case <-tk:
            if md.GetActiveMeta(true) == "" && md.serverMode {
                md.onSetActiveCallback(md.tryToBeActive())
            }
        case <-md.stopping:
            break LOOP
        case event := <-md.events():
            switch event.Type {
            case zk.EventNodeCreated:
            case zk.EventNodeDataChanged:
                md.loadNewActiveMeta()
            case zk.EventNodeDeleted:
                md.resetActiveMeta()
            default:
            }
            if md.GetActiveMeta(false) == "" && md.serverMode {
                md.onSetActiveCallback(md.tryToBeActive())
            }
            md.resetWatch()
        }
    }
    md.zkConn.Close()
    close(md.stopped)
    return nil
}
